package com.eksboot;

import com.eksframework.commons.json.EksJson;
import com.google.common.util.concurrent.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.util.StopWatch;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
public class CompletableFutureTest {

    public static void exec1() throws InterruptedException {
        log.info("线程："+Thread.currentThread().getId()+" 开始");
        TimeUnit.SECONDS.sleep(1);
        log.info("线程："+Thread.currentThread().getId()+" 结束");
    }

    public static void exec2() throws InterruptedException {
        log.info("线程："+Thread.currentThread().getId()+" 开始");
        TimeUnit.SECONDS.sleep(2);
        log.info("线程："+Thread.currentThread().getId()+" 结束");
    }

    public static void exec3() throws InterruptedException {
        log.info("线程："+Thread.currentThread().getId()+" 开始");
        TimeUnit.SECONDS.sleep(3);
        log.info("线程："+Thread.currentThread().getId()+" 结束");
    }

    public static void exec4() throws InterruptedException {
        log.info("线程："+Thread.currentThread().getId()+" 开始");
        TimeUnit.SECONDS.sleep(4);
        log.info("线程："+Thread.currentThread().getId()+" 结束");
    }

    public static void exec5() throws InterruptedException {
        log.info("线程："+Thread.currentThread().getId()+" 开始");
        int i = 1/0;
        log.info("{}", i);
        log.info("线程："+Thread.currentThread().getId()+" 结束");
    }

    public static void test1(){
        ExecutorService executor = Executors.newFixedThreadPool(2);
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            log.info("主线程：" + Thread.currentThread().getId() + " 开始");
            CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
                try {
                    exec1();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, executor);
            CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
                try {
                    exec2();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, executor);
            CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
                try {
                    exec3();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, executor);
            CompletableFuture<Void> future4 = CompletableFuture.runAsync(() -> {
                try {
                    exec4();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, executor);
            CompletableFuture<Void> future5 = CompletableFuture.runAsync(() -> {
                try {
                    exec5();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, executor);
            log.info("主线程：" + Thread.currentThread().getId() + " 进行中1");
            CompletableFuture<Void> allFuture = CompletableFuture.allOf(future1, future2, future3, future4, future5);
            log.info("主线程：" + Thread.currentThread().getId() + " 进行中2");
            allFuture.get();
            //allFuture.join();
        }catch (Exception ex){
            log.error("报错：", ex);
        }finally {
            stopWatch.stop();
            log.info("主线程：" + Thread.currentThread().getId() + " 结束，耗时：" + stopWatch.getTotalTimeMillis());
            executor.shutdown();
        }
    }

    public static void test2(){
        ExecutorService executor = Executors.newFixedThreadPool(2);
        ListeningExecutorService executorService = MoreExecutors.listeningDecorator(executor);
        StopWatch stopWatch = new StopWatch();
        stopWatch.start("start");
        try {
            log.info("主线程：" + Thread.currentThread().getId() + " 开始");

            ListenableFuture<Object> future1 = executorService.submit(() -> {
                try {
                    exec1();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "1";
            });
            ListenableFuture<Object> future2 = executorService.submit(() -> {
                try {
                    exec2();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "2";
            });
            ListenableFuture<Object> future3 = executorService.submit(() -> {
                try {
                    exec3();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "3";
            });
            ListenableFuture<Object> future4 = executorService.submit(() -> {
                try {
                    exec4();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "4";
            });
            ListenableFuture<Object> future5 = executorService.submit(() -> {
                try {
                    exec5();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "5";
            });
            log.info("主线程：" + Thread.currentThread().getId() + " 进行中1");
            ListenableFuture<List<Object>> listFuture = Futures.allAsList(future1, future2, future3, future4, future5);
            log.info("主线程：" + Thread.currentThread().getId() + " 进行中2");
            Futures.addCallback(listFuture, new FutureCallback<List<Object>>(){

                /**
                 * Invoked with the result of the {@code Future} computation when it is successful.
                 *
                 * @param result
                 */
                @Override
                public void onSuccess(@Nullable List<Object> result) {
                    log.info("onSuccess");
                    stopWatch.stop();
                    log.info("主线程：" + Thread.currentThread().getId() + " 结束，耗时：" + stopWatch.getTotalTimeMillis());
                    executorService.shutdown();
                    executor.shutdown();
                }

                /**
                 * Invoked when a {@code Future} computation fails or is canceled.
                 *
                 * <p>If the future's {@link Future#get() get} method throws an {@link ExecutionException}, then
                 * the cause is passed to this method. Any other thrown object is passed unaltered.
                 *
                 * @param t
                 */
                @SneakyThrows
                @Override
                public void onFailure(Throwable t) {
                    log.info("onFailure");
                    log.error("报错了", t);
                    stopWatch.stop();
                    log.info("主线程：" + Thread.currentThread().getId() + " 结束，耗时：" + stopWatch.getTotalTimeMillis());
                    executorService.shutdown();
                    executor.shutdown();
                    throw t;
                }
            }, executorService);


        }catch (Exception ex){
            log.error("报错：", ex);
        }finally {
            stopWatch.stop();
            stopWatch.start("finally");
            log.info("主线程：" + Thread.currentThread().getId() + " 结束");
        }
    }

    public static void test3(){
        log.info("主线程ID：{}", Thread.currentThread().getId());
        log.info("主线程组名称：{}", Thread.currentThread().getThreadGroup().getName());
        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<String> futureA = CompletableFuture.supplyAsync(() -> {
            log.info("A线程ID：{}", Thread.currentThread().getId());
            log.info("A线程组名称：{}", Thread.currentThread().getThreadGroup().getName());
            log.warn("我是A");
            return "我是A";
        }, executor);

        Future<String> futureB = CompletableFuture.supplyAsync(() -> {
            log.info("B线程ID：{}", Thread.currentThread().getId());
            log.warn("我是B");
            return "我是B";
        }, executor);
        try {
            System.out.println(futureA.get());
            System.out.println(futureB.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        CompletableFuture<String> futureC = CompletableFuture.supplyAsync(() -> {
            log.warn("我是C");
            return "我是C";
        }, executor);

        CompletableFuture<String> futureD = CompletableFuture.supplyAsync(() -> {
            log.warn("我是D");
            print();
            return "我是D";
        }, executor);

        CompletableFuture.allOf(futureC, futureD).join();
        System.out.println(futureC.join());
        System.out.println(futureD.join());

        System.out.println("===========================================");
        executor.shutdown();
    }

    public static void print(){
        System.out.println(EksJson.toJson(Thread.currentThread().getStackTrace()[2]));
    }

    public static void main(String[] args) {
        //test1();
        //test2();
        test3();
    }
}
